package com.wl.cloud.datax.common.utils;

import com.wl.cloud.core.exception.CustomException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateAclsResult;
import org.apache.kafka.clients.admin.DeleteAclsResult;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.acl.*;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @Author: panzhichun
 * @Date: 2024/2/22 13:50
 * @Description:
 * @Version: 1.0
 */
@Slf4j
public class CustomKafkaAclsUtils {

    // 利用已知信息拼装
    public static AclBinding createAclBinding(String resourceType, String resourceName, String username, String host, String operation) {
        ResourcePattern resource = new ResourcePattern(getResourceType(resourceType), resourceName, PatternType.LITERAL);
        AccessControlEntry accessControlEntry = new AccessControlEntry("User:" + username, host, getOperation(operation), AclPermissionType.ALLOW);

        return new AclBinding(resource, accessControlEntry);
    }


    /**
     * 添加权限
     *
     */
    public static void addAclAuth(AdminClient adminClient, List<AclBinding> aclBindingList) {
        CreateAclsResult createAclsResult = adminClient.createAcls(aclBindingList);
        for (Map.Entry<AclBinding, KafkaFuture<Void>> e : createAclsResult.values().entrySet()) {
            KafkaFuture<Void> future = e.getValue();
            try {
                future.get();
                boolean success = !future.isCompletedExceptionally();
                if (success) {
                    log.info("创建权限成功");
                }
            } catch (Exception exc) {
                log.warn("创建权限失败，错误信息：{}", exc.getMessage());
                throw new CustomException(exc);
            }
        }
    }

    /**
     * 删除权限
     *
     */
    public static void deleteAclAuth(AdminClient adminClient, List<AclBinding> aclBindingList) {
        List<AclBindingFilter> aclBindingFilterList = new ArrayList<>();
        aclBindingList.forEach(item -> {
            aclBindingFilterList.add(item.toFilter());
        });
        DeleteAclsResult deleteAclsResult = adminClient.deleteAcls(aclBindingFilterList);
        for (Map.Entry<AclBindingFilter, KafkaFuture<DeleteAclsResult.FilterResults>> e : deleteAclsResult.values().entrySet()) {
            KafkaFuture<DeleteAclsResult.FilterResults> future = e.getValue();
            try {
                future.get();
                boolean success = !future.isCompletedExceptionally();
                if (success) {
                    log.info("删除权限成功");
                }
            } catch (Exception exc) {
                log.warn("删除权限失败，错误信息：{}", exc.getMessage());
                throw new CustomException(exc);
            }
        }
    }

    private static AclOperation getOperation(String operation) {
        AclOperation aclOperation = null;
        String operation1 = operation.toLowerCase();
        switch (operation1) {
            case "create":
                aclOperation = AclOperation.CREATE;
                break;
            case "write":
                aclOperation = AclOperation.WRITE;
                break;
            case "read":
                aclOperation = AclOperation.READ;
                break;
            case "alter":
                aclOperation = AclOperation.ALTER;
                break;
            case "describe":
                aclOperation = AclOperation.DESCRIBE;
                break;
            case "all":
                aclOperation = AclOperation.ALL;
                break;
            default:
                throw new CustomException("暂不支持的操作权限:" + operation);
        }
        return aclOperation;
    }

    private static ResourceType getResourceType(String type) {
        ResourceType resourceType = null;
        String type1 = type.toLowerCase();
        switch (type1) {
            case "group":
                resourceType = ResourceType.GROUP;
                break;
            case "topic":
                resourceType = ResourceType.TOPIC;
                break;
            case "cluster":
                resourceType = ResourceType.CLUSTER;
                break;
            case "transactionalid":
                resourceType = ResourceType.TRANSACTIONAL_ID;
                break;
            default:
                throw new CustomException("不支持的资源类型:" + type);
        }
        return resourceType;
    }

}
